package cn.jn.lhm.netty.part08.chat.rpc;

import cn.jn.lhm.netty.part08.chat.handler.RpcHandler;
import cn.jn.lhm.netty.part08.chat.message.RpcRequestMessage;
import cn.jn.lhm.netty.part08.chat.protocol.ProtocolFrameDecoder;
import cn.jn.lhm.netty.part08.chat.protocol.SequenceIdGenerator;
import cn.jn.lhm.netty.part08.chat.protocol.SharableMessageCodec;
import cn.jn.lhm.netty.part08.chat.service.HelloService;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;
import io.netty.util.concurrent.DefaultPromise;
import lombok.extern.slf4j.Slf4j;

import java.lang.reflect.Proxy;

/**
 * @author lhm 2753914576@qq.com
 * @time 2023/11/2 - 23:11
 **/
@Slf4j
public class RpcClientManager {


    public static void main(String[] args) {
        HelloService helloServer = getProxy(HelloService.class);
        helloServer.sayHello("lisi");
        helloServer.sayHello("zhangsan");
        helloServer.sayHello("王五");
    }

    @SuppressWarnings("all")
    public static <T> T getProxy(Class<T> clazz) {
        return (T) Proxy.newProxyInstance(clazz.getClassLoader(), new Class[]{clazz}, (proxy, method, args) -> {
            // 封装为 RPC 消息
            int seqId = SequenceIdGenerator.nextId();
            RpcRequestMessage requestMessage = new RpcRequestMessage(seqId, clazz.getName(), method.getName(), method.getReturnType(), method.getParameterTypes(), args);
            // 执行 RPC 调用
            getRpcClientChannel().writeAndFlush(requestMessage);
            // 返回调用结果
            // 这里 evertLoop 就是指定Promise异步处理时使用的线程
            DefaultPromise<Object> promise = new DefaultPromise<>(getRpcClientChannel().eventLoop());
            RpcHandler.RESULT_MAP.put(seqId, promise);

            // 等待返回结果
            promise.await();

            if (promise.isSuccess()) {
                return promise.get();
            }
            throw new RuntimeException(promise.cause());
        });
    }

    private static Channel channel;

    private static final Object LOCK = new Object();

    public static Channel getRpcClientChannel() {
        if (channel != null) {
            return channel;
        }
        synchronized (LOCK) {
            if (channel != null) {
                return channel;
            }
            initChannel();
            return channel;
        }
    }

    private static void initChannel() {
        NioEventLoopGroup worker = new NioEventLoopGroup();
        Bootstrap bootstrap = new Bootstrap();
        bootstrap.channel(NioSocketChannel.class);
        bootstrap.group(worker);
        bootstrap.handler(new ChannelInitializer<NioSocketChannel>() {
            @Override
            protected void initChannel(NioSocketChannel ch) throws Exception {
                ch.pipeline().addLast(new ProtocolFrameDecoder());
                ch.pipeline().addLast(new LoggingHandler(LogLevel.DEBUG));
                ch.pipeline().addLast(new SharableMessageCodec());
                ch.pipeline().addLast(new RpcHandler());
            }
        });
        ChannelFuture future = bootstrap.connect("localhost", 8080);
        try {
            channel = future.sync().channel();
            channel.closeFuture().addListener(closeFuture -> worker.shutdownGracefully());
        } catch (Exception e) {
            worker.shutdownGracefully();
            throw new RuntimeException(e);
        }
    }

}
